package com.im.server.init;

import com.google.protobuf.MessageLite;
import com.google.protobuf.MessageLiteOrBuilder;
import com.im.common.Constants;
import com.im.server.handle.ImServerHandle;
import com.im.server.protocol.MessageProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.stream.ChunkedWriteHandler;

import java.util.List;

import static io.netty.buffer.Unpooled.wrappedBuffer;

/**
 * @author gongym
 * @date 2019/1/9 13:08
 */
public class ImServerInitializer extends ChannelInitializer<Channel> {
  private final ImServerHandle imServerHandle = new ImServerHandle();

  @Override
  protected void initChannel(Channel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    // HTTP请求的解码和编码
    pipeline.addLast(new HttpServerCodec());
    // 把多个消息转换为一个单一的FullHttpRequest或是FullHttpResponse，
    // 原因是HTTP解码器会在每个HTTP消息中生成多个消息对象HttpRequest/HttpResponse,HttpContent,LastHttpContent
    pipeline.addLast(new HttpObjectAggregator(Constants.MAX_AGGREGATED_CONTENT_LENGTH));
    // 主要用于处理大数据流，比如一个1G大小的文件如果你直接传输肯定会撑暴jvm内存的; 增加之后就不用考虑这个问题了
    pipeline.addLast(new ChunkedWriteHandler());
    // WebSocket数据压缩
    pipeline.addLast(new WebSocketServerCompressionHandler());
    // 协议包长度限制
    pipeline.addLast(
        new WebSocketServerProtocolHandler("/chat", null, true, Constants.MAX_FRAME_LENGTH));
    // 防止粘包问题
    pipeline.addLast(new ProtobufVarint32FrameDecoder());
    pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
    // 协议包解码
    pipeline.addLast(
        new MessageToMessageDecoder<WebSocketFrame>() {
          @Override
          protected void decode(
              ChannelHandlerContext ctx, WebSocketFrame frame, List<Object> objs) {
            ByteBuf buf = frame.content();
            objs.add(buf);
            buf.retain();
          }
        });
    // 协议包编码
    pipeline.addLast(
        new MessageToMessageEncoder<MessageLiteOrBuilder>() {
          @Override
          protected void encode(
              ChannelHandlerContext ctx, MessageLiteOrBuilder msg, List<Object> out) {
            ByteBuf result = null;
            if (msg instanceof MessageLite) {
              result = wrappedBuffer(((MessageLite) msg).toByteArray());
            }
            if (msg instanceof MessageLite.Builder) {
              result = wrappedBuffer(((MessageLite.Builder) msg).build().toByteArray());
            }
            // ==== 上面代码片段是拷贝自TCP ProtoBufEncoder 源码 ====
            // 然后下面再转成websocket二进制流，因为客户端不能直接解析protoBuf编码生成的
            assert result != null;
            WebSocketFrame frame = new BinaryWebSocketFrame(result);
            out.add(frame);
          }
        });
    pipeline.addLast(new ProtobufDecoder(MessageProto.MessageProtocol.getDefaultInstance()));
    pipeline.addLast(imServerHandle);
  }
}
